package com.unipay.oss;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.Properties;

/**
 * @Project: yitong_parent
 * @Author: Chenmy
 * @Time: 2023-10-20 11:19:40
 * @Desc: TODO
 */

public class KfConsumer {
    public static void main(String[] args) {
        Properties prop = new Properties();

        prop.put("bootstrap.servers", "172.18.177.141:9092,172.18.177.144:9092,172.18.177.143:9092");
        prop.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");

        prop.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put("group.id", "con-1");
        prop.put("auto.offset.reset", "latest");
        //自动提交偏移量
        prop.put("auto.commit.intervals.ms", "true");
        //自动提交时间
        prop.put("auto.commit.interval.ms", "1000");
        KafkaConsumer<String, String> consumer = new  KafkaConsumer<>(prop);
        ArrayList<String> topics = new ArrayList<>();
        //可以订阅多个消息
        topics.add("cmyTest");
        consumer.subscribe(topics);
        while (true) {
            //ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(20));
            ConsumerRecords<String, String> poll = consumer.poll(20);
            for (ConsumerRecord<String, String> consumerRecord : poll) {
                System.out.println(consumerRecord);
            }
        }
    }
}
